package com.hjc.project.util;

import com.hjc.project.util.diruptor.OrderEventProducer;
import com.hjc.project.util.disruptor.QueryEventProducer;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;

import java.io.IOException;

public class EsScrollQueryUtils {

    QueryEventProducer producer;

    public void scrollQueryTest() throws IOException {
        //        1. 创建查询对象
        String index = "third_event_list_202104";
        String type = "log";
        SearchRequest searchRequest = new SearchRequest(index);//指定索引
        searchRequest.types(type);//指定类型
        searchRequest.scroll(TimeValue.timeValueMinutes(1L));//指定存在内存的时长为1分钟
//    2. 封装查询条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.sort("date", SortOrder.DESC);
        searchSourceBuilder.size(2);
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        searchRequest.source(searchSourceBuilder);


        //        3.执行查询
        // client执行
        HttpHost httpHost = new HttpHost("192.168.43.30", 9200);
        RestClientBuilder restClientBuilder = RestClient.builder(httpHost);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClientBuilder);

        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = searchResponse.getScrollId();
        System.out.println(scrollId);//获取scorllId


//        4.获取数据
        SearchHit[] hits = searchResponse.getHits().getHits();
        for(SearchHit searchHit : hits){
            System.out.println(searchHit);
        }

        while (true){
            //创建SearchScrollRequest对象
            SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
            searchScrollRequest.scroll(TimeValue.timeValueMinutes(1L));//设置1分钟
            SearchResponse scroll = restHighLevelClient.scroll(searchScrollRequest, RequestOptions.DEFAULT);
            SearchHit[] hits1 = scroll.getHits().getHits();
            if(hits1 != null && hits1.length > 0){
                System.out.println("------------下一页--------------");
                for(SearchHit searchHit : hits1){
                    System.out.println(searchHit);
                    producer.sendData(searchHit.getSourceAsString());
                }

            }else {
                System.out.println("------------结束--------------");
                break;
            }
        }

        //删除ScrollId
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        System.out.println("删除scroll"  + clearScrollResponse);
    }
}
